package org.idea.qiyu.framework.canal.producer;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.Message;
import org.idea.qiyu.framework.canal.handle.CanalMessageQueueHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;

import java.util.Set;

/**
 * @Author linhao
 * @Date created in 2:37 下午 2021/6/20
 */
public class CanalMessageProducer implements Runnable {

    private static final Logger LOGGER = LoggerFactory.getLogger(CanalMessageProducer.class);

    private Set<CanalConnector> canalConnectorList;

    private CanalMessageQueueHandler canalMessageQueueHandler;

    public CanalMessageProducer(Set<CanalConnector> canalConnectors, CanalMessageQueueHandler canalMessageQueueHandler) {
        this.canalConnectorList = canalConnectors;
        this.canalMessageQueueHandler = canalMessageQueueHandler;
    }

    @Override
    public void run() {
        if (CollectionUtils.isEmpty(canalConnectorList)) {
            return;
        }
        LOGGER.info("[CanalMessageProducer] is started!");
        while (true) {
            for (CanalConnector canalConnector : canalConnectorList) {
                Message message = canalConnector.getWithoutAck(100);
                if(message==null || message.getId()<0){
                    continue;
                }
                long batchId = message.getId();
                canalMessageQueueHandler.addMessage(message);
                canalConnector.ack(batchId);
            }
        }
    }
}
